
#include "./cluster/driver.h"
#include "./cluster/worker.h"

#include <fstream>

#define DRIVER_PROP "drivers"
#define WORKER_PROP "workers"
#define REPLICATION_PROP "replication"

static int __port_ = -1;

std::pair<std::string, std::string> parse_properties_line(const std::string & line) {
    auto eidx = line.find_first_of('=');
    auto i = eidx - 1, j = eidx + 1;
    while (i >= 0 && line[i] == ' ') --i;
    while (j < line.size() && line[j] == ' ') ++j;
    size_t ii = 0, jj = line.size() - 1;
    while (ii < line.size() && line[ii] == ' ') ++ii;
    while (jj >= 0 && line[jj] == ' ') --jj;
    return { line.substr(ii, i - ii + 1), line.substr(j, jj - j + 1) };
}

std::unordered_map<std::string, std::string> parse_properties(const char * conf_path) {
    std::vector<std::string> props;
    std::ifstream fin(conf_path);
    if (!fin) EXIT_ERR("cannot open cluster_info.properties");
    std::unordered_map<std::string, std::string> ans;
    std::string line;
    while (std::getline(fin, line)) {
        std::pair<std::string, std::string> prop = parse_properties_line(line);
        ans[prop.first] = prop.second;
    }
    fin.close();
    return ans;
}

sockaddr_in gen_sockaddr_by_hp(const std::string & hp) {
    auto midx = hp.find_first_of(':');
    std::string host_s = hp.substr(0, midx);
    std::string port_s = hp.substr(midx + 1);
    const char * host = host_s.c_str();
    int port = std::stoi(port_s);
    sockaddr_in ans { };
    memset(&ans, 0, sizeof(ans));
    ans.sin_family = AF_INET;
    ans.sin_port = htons(port);
    ans.sin_addr.s_addr = inet_addr(host);
    return ans;
}

std::vector<sockaddr_in> gen_sock_addrs(const std::string & addr_str) {
    std::vector<std::string> addrs = lukv::split_by_char(addr_str, ',');
    std::vector<sockaddr_in> ans;
    for (auto & addr: addrs) {
        ans.push_back(gen_sockaddr_by_hp(addr));
    }
    return ans;
}

void init_cluster_info(const char * conf_path) {
    std::unordered_map<std::string, std::string> props = parse_properties(conf_path);
    std::string driver_hps = props[DRIVER_PROP];
    std::string worker_hps = props[WORKER_PROP];

    int rep = std::stoi(props[REPLICATION_PROP]);
    std::vector<sockaddr_in> driver_addrs = gen_sock_addrs(driver_hps);
    std::vector<sockaddr_in> worker_addrs = gen_sock_addrs(worker_hps);

    lukv::manager::set_driver_addrs(driver_addrs);
    lukv::manager::set_worker_addrs(worker_addrs);
    lukv::manager::set_replication(rep);
}

std::string get_ip_addr() {
    char hostname[256];

    gethostname(hostname, sizeof(hostname));

    hostent * host = gethostbyname(hostname);

    auto ** addr_list = (in_addr **)host->h_addr_list;

    return inet_ntoa(*addr_list[0]);
}

// e.g.: ssh root@xxx.xxx.xxx.xxx "nohup $LUKV_HOME/bin/lukv_cluster  2>&1 &"
std::string gen_notify_cmd(sockaddr_in & _addr) {
    const char * _h = inet_ntoa(_addr.sin_addr);
    std::string _p = std::to_string(ntohs(_addr.sin_port));
    std::string ans;
    ans.append("ssh root@");
    ans.append(_h);
    ans.append(" \"nohup /usr/develop/cpp/projects/lukv/cmake-build-debug/lukv_cluster " + _p + " 2>&1 1>&1 &\"");
    return ans;
}

int notify_oth_nodes_start() {
    for (int i = 0; i < lukv::manager::driver_addrs.size(); ++i) {
        if (i == 0) continue;
        std::string _cmd = gen_notify_cmd(lukv::manager::driver_addrs[i]);
        lukv::println("ready to notify driver ", i, " to start, cmd: ", _cmd);
        system(_cmd.c_str());
        lukv::println("notified driver ", i, " to start...");
    }
    for (int i = 0; i < lukv::manager::worker_addrs.size(); ++i) {
        std::string _cmd = gen_notify_cmd(lukv::manager::worker_addrs[i]);
        lukv::println("ready to notify worker ", i, " to start, cmd: ", _cmd);
        system(_cmd.c_str());
        lukv::println("notified worker ", i, " to start...");
    }
    return 0;
}

int start_cluster(const char * conf_path) {
    init_cluster_info(conf_path);
    std::string local_ip = get_ip_addr();
    for (auto i = 0; i < lukv::manager::driver_addrs.size(); ++i) {
        if (local_ip == inet_ntoa(lukv::manager::driver_addrs[i].sin_addr) && __port_ == ntohs(lukv::manager::driver_addrs[i].sin_port)) {
            if (i == 0) { // the first driver (master) needs to notify the other nodes to start.
                if (notify_oth_nodes_start() < 0) return -2;
            }
            lukv::driver drv(i);
            lukv::println("starting driver: ", local_ip, ":", __port_);
            return drv.start();
        }
    }
    for (auto i = 0; i < lukv::manager::worker_addrs.size(); ++i) {
        if (local_ip == inet_ntoa(lukv::manager::worker_addrs[i].sin_addr) && __port_ == ntohs(lukv::manager::worker_addrs[i].sin_port)) {
            lukv::worker wkr(i);
            lukv::println("starting worker: ", local_ip, ":", __port_);
            return wkr.start();
        }
    }
    return -1;
}

int main(int argc, char** argv) {

    if (argc < 2) EXIT_ERR("too less argument!");

    __port_ = std::stoi(argv[1]);
    
    const char * conf_path = "/usr/develop/cpp/projects/lukv/config/cluster_info.properties";
    if (start_cluster(conf_path) < 0) EXIT_ERR(strerror(errno));

    return 0;
}

